package com.ruoyi.passenger.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * @author : 李远程
 * @date : 2024-05-25 11:25
 * @description :
 **/
@Configuration
public class KafKaConfig {

    @Bean(name = "kafkaProducer")
    public KafkaProducer<String, String> KafkaProducer() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.19.29.83:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.springframework.kafka.support.serializer.JsonSerializer");
        config.put(ProducerConfig.CLIENT_ID_CONFIG, "ProducerTranscationnalExample");
        config.put(ProducerConfig.ACKS_CONFIG, "all");//保证可靠all -1
        KafkaProducer<String, String> producer = new KafkaProducer<>(config);
        return producer;
    }

    @Bean(name = "kafkaConsumer")
    public KafkaConsumer<String, String> kafkaConsumer() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.19.29.83:9092");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.springframework.kafka.support.serializer.JsonDeserializer");
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "2109A-group1-2");
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);//自动提交选项
        /*Properties properties = new Properties();
        properties.put("spring.json.trusted.packages", "*");*/
        config.put("spring.json.trusted.packages", "*");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(config);
        return consumer;
    }


    @Bean
    public NewTopic newTopic() {
        return new NewTopic("my-2109A-topic", 3, (short) 1);
    }

}
